package com.zx.mes.rocketmq.hyl.quickstart;

import com.zx.mes.rocketmq.hyl.constant.Const;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

/**
 * rocket mq async producer
 * 异步生产者
 *
 * @author 华云龙
 * @date 2018-11-27
 */
public class ProducerAsync {
    private static final int INT = 5;
    private static final String TEST_QUICK_TOPIC = "test_quick_topic";

    public static void send() throws MQClientException, RemotingException, InterruptedException {
        String producerGroup = "test_quick_producer_name";
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(Const.NAMESRV_ADDR_SINGLE);
        producer.start();

        for (int i = 0; i <INT ; i++) {

            Message message = new Message(TEST_QUICK_TOPIC, "Top B", "keyB" + i, ("神的世界一般人不懂!" + i).getBytes());

            /// 异步消费
            /// rabbit mq百分百投递
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    /// 成功修数据状态
                    System.out.println("发送成功,状态:"+sendResult.getSendStatus()+",msgId:"+sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable e) {
                    /// 做补偿
                    System.err.println("发送失败!,失败日志:"+e.getMessage());
                }
            });
        }

        Thread.sleep(2000);

        /// 关闭生产者
        producer.shutdown();
    }
}
